package indi.zjh.dp.mng;

import indi.zjh.dp.db.Configuration;
import indi.zjh.dp.startup.Constant;
import indi.zjh.dp.startup.ServerStatus;
import indi.zjh.dp.util.JSONUtil;
import indi.zjh.dp.util.SerializeUtil;
import indi.zjh.dp.util.XMLUtil;

import java.io.File;




import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException.NodeExistsException;
import org.dom4j.DocumentException;

import com.netflix.curator.RetryPolicy;
import com.netflix.curator.framework.CuratorFramework;
import com.netflix.curator.framework.CuratorFrameworkFactory;
import com.netflix.curator.framework.recipes.cache.NodeCache;
import com.netflix.curator.framework.recipes.cache.NodeCacheListener;
import com.netflix.curator.retry.ExponentialBackoffRetry;

public class Manager {
	private static Logger logger = Logger.getLogger(Manager.class);
	
	private static Manager instance = null;
//	public static final String CONF_FILE = "conf/server.xml";
//	private String hostList = ""; // zookeeper服务器列表
//	private int sessionTimeout = 5000; // zookeeper会话超时
//	private int connectTimeout = 5000; // zookeeper连接超时
//	private int reconnectTimeout = 1000; // 重连超时
//	private int reconnectTryCount = 3; // 重连次数
	private ServerStatus busiServerStatus = null;
	
	class MonitorThread extends Thread
	{
		private Logger logger = null;
		private Manager manager = null;
		private long startTime = 0;
		private long lastUpdateTime = 0;
		public MonitorThread(Manager m, Logger logger)
		{
			this.logger = logger;
			this.manager = m;
			this.startTime = (int)(System.currentTimeMillis()/1000);
		}
		
		public void run()
		{
			try
			{
				while(true)
				{
					Thread.sleep(10000);
					ServerStatus status = this.manager.getBusinessServerStatus();
					if(status == null)
					{
						this.lastUpdateTime = this.startTime;
					}
					else
					{
						this.lastUpdateTime = status.getUpdateTime();
					}
					long currTime = (int)(System.currentTimeMillis()/1000);
					this.logger.info("管理节点检查业务节点心跳:"+this.lastUpdateTime);
					if(currTime - this.lastUpdateTime > 10)
					{
						logger.error("业务节点心跳超时");
					}
				}
			}
			catch(Exception e)
			{
				e.printStackTrace();
			}
		}
	}
	
	public static Manager getInstance()
	{
		if(instance == null)
		{
			instance = new Manager();
		}
		return instance;
	}
	
	public Manager()
	{
		
	}
	
	public void init()
	{
		CuratorFramework client = null;
		try
		{
			// 服务器启动
			logger.info("服务器节点启动");
			String hostList = Configuration.getInstance().getZkConf().host;
			logger.info("连接zookeeper服务器列表:"+hostList);
			int sessionTimeout = Configuration.getInstance().getZkConf().sessionTimeout;
			int connectTimeout = Configuration.getInstance().getZkConf().connectTimeout;
			logger.info("会话超时:"+sessionTimeout+", 连接超时:"+connectTimeout);
			int reconnectTimeout = Configuration.getInstance().getZkConf().reconnectTimeout;
			int reconnectTryCount = Configuration.getInstance().getZkConf().reconnectTryCount;
			logger.info("重连超时:"+reconnectTimeout+", 重连次数:"+reconnectTryCount);
			
			// 重连策略
			RetryPolicy retryPolicy = new ExponentialBackoffRetry(reconnectTimeout, reconnectTryCount);
			
			// 连接
			client = CuratorFrameworkFactory.newClient(hostList, sessionTimeout, connectTimeout, retryPolicy);
			client.start();
			
			// 注册状态变化事件监听
			client.getConnectionStateListenable().addListener(new ServerConnectionStateListener());
			
			// 创建/busi_node目录
			try
			{
				client.create().withMode(CreateMode.PERSISTENT).forPath("/busi_node");
			}
			catch(NodeExistsException e)
			{
				
			}
			
			// 创建/manager_node目录
			try
			{
				client.create().withMode(CreateMode.PERSISTENT).forPath("/manager_node");
			}
			catch(NodeExistsException e)
			{
				
			}
			
			String managerName = client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath("/manager_node/manager");

			String nodeName = client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath("/busi_node/node");

			ServerStatus dump = ServerStatus.getInstance();
			client.setData().forPath(nodeName, JSONUtil.toString(dump).getBytes());
			
			
			NodeCache cache = new NodeCache(client, nodeName);
			cache.start();
			cache.getListenable().addListener(new NodeCacheListener(){

				@Override
				public void nodeChanged() throws Exception {
					// TODO Auto-generated method stub
					byte[] data = cache.getCurrentData().getData();
					logger.info("管理节点监听到业务节点的心跳内容:"+new String(data));
					ServerStatus status = (ServerStatus)JSONUtil.toObject(new String(data), ServerStatus.class); //(ServerStatus)SerializeUtil.unserialize(data, ServerStatus.class);
					Manager.this.updateBusinessServerStatus(status);
				}
				
			});
			
			
			System.out.println("./startNode.sh "+managerName+" "+nodeName);
			Runtime.getRuntime().exec("./startNode.sh "+managerName+" "+nodeName);
			
			
			new MonitorThread(this, logger).start();
			
			Thread.sleep(Integer.MAX_VALUE);
		}

		catch(Exception e)
		{
			e.printStackTrace();
			System.exit(1);
		}
	}
	
	public ServerStatus getBusinessServerStatus()
	{
		return this.busiServerStatus;
	}
	
	public void updateBusinessServerStatus(ServerStatus status)
	{
		this.busiServerStatus = status;
	}
}
